package cloudy.bolt;

import cloudy.tools.DateFmt;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.BasicOutputCollector;
import org.apache.storm.topology.IBasicBolt;
import org.apache.storm.topology.IRichBolt;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.util.Map;

/**
 * @Name cloudy.bolt.AreaFilterBolt
 * @Description
 * @Author Elwyn
 * @Version 2017/9/29
 * @Copyright 上海云辰信息科技有限公司
 **/
public class AreaFilterBolt implements IBasicBolt {

	@Override
	public void prepare(Map map, TopologyContext topologyContext) {

	}

	@Override
	public void execute(Tuple tuple, BasicOutputCollector basicOutputCollector) {
		String order = tuple.getString(0);
		if (order != null) {
			String[] orderArr = order.split("\\t");
			basicOutputCollector.emit(new Values(orderArr[3], orderArr[1], DateFmt.getCountDate(orderArr[2], DateFmt.date_short)));
		}
	}

	@Override
	public void cleanup() {

	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
		outputFieldsDeclarer.declare(new Fields("area_id", "order_amt", "order_date"));
	}

	@Override
	public Map<String, Object> getComponentConfiguration() {
		return null;
	}
}
